package edu.kit.ksri.db.mturk.initializer;

import java.io.File;
import java.text.DecimalFormat;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import com.amazonaws.mturk.requester.HIT;

import edu.kit.ksri.db.mturk.connectionhandlers.AbstractHITManager;
import edu.kit.ksri.db.mturk.plugins.correctness.AbstractQualityControlImplementation;
import edu.kit.ksri.db.mturk.plugins.display.BrowserDisplay;
import edu.kit.ksri.db.mturk.plugins.performance.TimeConstraintsManager;
import edu.kit.ksri.db.mturk.qualifications.QualificationManager;
import edu.kit.ksri.db.mturk.utils.messaging.HITResultFetcher;
import edu.kit.ksri.db.mturk.utils.messaging.ResultOutputCollector;
import edu.kit.ksri.db.mturk.var.ConnectionMode;
import edu.kit.ksri.db.mturk.var.*;
import edu.kit.ksri.db.mturk.workerpools.PoolManager;
import edu.kit.ksri.db.mturk.workerpools.WorkerPool;

/**
 * Copyright (c) 2010-2011 Karlsruhe Institute of Technology (KIT), Germany
 * 
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 * 
 * --------------------------------------------------------------------------
 * 
 * Software: CSP/WMV tool for dynamic quality management of microtasks
 * http://www.ksri.kit.edu/Upload/Users/PeopleClouds/
 * HCOMP2011_An_extendable_toolkit_for_managing_quality_of_human
 * -based_eServices.pdf
 * 
 * Organization: Karlsruhe Institute of Technology (KIT), Germany
 * http://www.kit.edu
 * 
 * Project: PeopleClouds project of the Karlsruhe Service Research
 * Institute(KSRI) http://www.ksri.kit.edu/Default.aspx?PageId=818
 * 
 * Contributing authors: David Bermbach, Sandra Rath, Pascal Wichmann, Robert
 * Kern
 */
public class StandaloneInitializer {

	private static final Logger basiclog = Logger.getLogger("basic."
			+ StandaloneInitializer.class);

	private static final Logger perflog = Logger.getLogger("perf."
			+ StandaloneInitializer.class);

	private static String[] steps = { "Step 0: Creating worker pools.",
			"Step 1: Importing worker information.",
			"Step 2: Initializing Qualification Manager.",
			"Step 3: Publishing tasks on pServices platform.",
			"Step 4: Starting HITResultFetcher.",
			"Step 5: Initializing Correctness Plugin.",
			"Step 6: Initializing Performance Plugin.",
			"Step 7: Waiting for results.",
			"Step 8: Finalizing and persisting state." };

	private static final DecimalFormat percentageFormat = new DecimalFormat(
			"#,###,##0.0");

	/**
	 * can be set externally to terminate step 7 earlier than planned as it
	 * might sometimes be an infinite loop. Is set to false before loop, so end
	 * of start up phase can be identified.
	 */
	private static boolean quit = false;

	/** is set to false as soon as all components are running */
	private static boolean starting = true;

	public static void main(String[] args) {
		// Delete old log
		File file = new File("mturkTool.log");
		if (file.exists()) {
			file.delete();
		}

		// Start new log
		PropertyConfigurator.configure("log4j.properties");
		basiclog.info("Initializing...");
		basiclog.info(StaticInformation.scenario.getParameters());
		new Thread(new QuitFrame()).start();
		boolean isPerfSim = false;
		boolean isCorrSim = false;
		boolean isSim = false;
		boolean isCorrSimInput = false;
		String workerFile = StaticInformation.workerFileIn;
		if (StaticInformation.scenario.getConnectionMode() == ConnectionMode.PERFSIM) {
			steps[3] = "Step 3: Initializing PerfSimMturk.";
			isPerfSim = true;
			workerFile = StaticInformation.workerSimInputFile;
		} else if (StaticInformation.scenario.getConnectionMode() == ConnectionMode.CORRSIM) {
			steps[3] = "Step 3: Initializing CorrSimMturk.";
			isCorrSim = true;
			workerFile = StaticInformation.workerSimInputFile;
		} else if (StaticInformation.scenario.getConnectionMode() == ConnectionMode.CORRSIMINPUT) {
			steps[3] = "Step 3: Initializing CorrSimMturk - input mode.";
			isCorrSimInput = true;
		} else if (StaticInformation.scenario.getConnectionMode() == ConnectionMode.SIM) {
			steps[3] = "Step 3: Initializing SimMturk.";
			isSim = true;
			workerFile = StaticInformation.workerSimInputFile;
		}
		int stepCounter = 0;
		try {
			// create worker pools
			basiclog.info(steps[stepCounter++]);
			initWorkerPools();
			// import workers and assign respective qualification values
			basiclog.info(steps[stepCounter++]);
			PoolManager.getInstance().importWorkers(workerFile, true);
			// start Qualification Manager
			QualificationManager qm = new QualificationManager();
			Thread qmthread = new Thread(qm);
			if (isPerfSim) {
				// don't start
				basiclog.info(steps[stepCounter++] + " - skipped.");
			} else if (isCorrSim) {
				// don't start
				basiclog.info(steps[stepCounter++] + " - skipped.");
			} else if (isSim) {
				// don't start
				basiclog.info(steps[stepCounter++] + " - skipped.");
			} else {
				// start for live mode and for corrsiminput mode
				basiclog.info(steps[stepCounter++]);
				qmthread.start();
			}

			// check if there already some HITs for this HIT type on MTurk. If
			// yes: continue
			basiclog.info(steps[stepCounter++]);
			HIT[] hits = null;

			if (!isPerfSim && !isCorrSim && !isSim) {
				hits = AbstractHITManager.getInstance().continuePreviousRun();
			}

			if (hits == null) {
				// New run, no previous run possible to continue
				// start simulation or create hits
				if (isPerfSim) {
					hits = AbstractHITManager.getInstance().createHITs(
							StaticInformation.simMTurkInputFile, null, null);
				} else if (isCorrSim) {
					hits = AbstractHITManager.getInstance().createHITs(
							StaticInformation.simMTurkInputDirectory, null,
							null);
				} else if (isSim) {
					hits = AbstractHITManager.getInstance().createHITs(
							StaticInformation.simMTurkInputDirectory, null,
							null);
				} else {// for live mode and for corrsiminput mode
					hits = AbstractHITManager.getInstance().createHITs(
							StaticInformation.inputFile,
							StaticInformation.questionFile,
							StaticInformation.propertiesFile);
				}
				if (hits == null)
					throw new NullPointerException();
			}

			if (StaticInformation.isMonitoring)
				BrowserDisplay.initialize(hits);

			// start hitresultfetcher
			basiclog.info(steps[stepCounter++]);
			Thread hitresultfetcherthread = new Thread(
					HITResultFetcher.getInstance());
			hitresultfetcherthread.start();
			// start csp1 and wmv
			AbstractQualityControlImplementation correctnessPlugin = StaticInformation.scenario
					.getCorrectnessPluginClass().newInstance();
			Thread correctnessthread = new Thread(correctnessPlugin);
			basiclog.info(steps[stepCounter++]);
			correctnessthread.start();

			// start timeconstraintsmanager
			Thread timeconstraintsthread = null;
			if (!isCorrSim && !isCorrSimInput && !isSim) {
				basiclog.info(steps[stepCounter++]);
				TimeConstraintsManager performancePlugin = new TimeConstraintsManager(
						StaticInformation.scenario.getDeadline().getTime(),
						StaticInformation.scenario.getEstimatedTaskDuration());
				timeconstraintsthread = new Thread(performancePlugin);
				timeconstraintsthread.start();
			} else {
				basiclog.info(steps[stepCounter++] + " - skipped.");
			}
			// wait for results and monitor when to "close" the application
			basiclog.info(steps[stepCounter++]);
			int waitLongerCounter = 0;
			ResultOutputCollector r = ResultOutputCollector.getInstance();
			starting = false;
			while (true) {
				if (isQuit()) {
					basiclog.warn("Terminating components earlier than planned.");
					// stop hitresultfetcher
					basiclog.info("Stopping HITResultFetcher.");
					HITResultFetcher.getInstance().setRunning(false);
					perflog.info("Waiting for HITResultFetcher to terminate...");
					int counter = 0;
					while (!HITResultFetcher.getInstance().isTerminated()) {
						perflog.info("Still waiting for HITResultFetcher to terminate...");
						Thread.sleep(1000);
						counter++;
						if (counter >= 10) {
							basiclog.info("Forcing HITResultFetcher to terminate...");
							hitresultfetcherthread.interrupt();
							HITResultFetcher.getInstance().setTerminated(true);
							break;
						}
					}
					perflog.info("HITResultFetcher terminated.");
					// stop Qualification Manager
					basiclog.info("Stopping Qualification Manager.");
					QualificationManager.setRunning(false);
					perflog.info("Waiting for Qualification Manager to terminate...");
					counter = 0;
					while (!QualificationManager.isTerminated()) {
						perflog.info("Still waiting for Qualification Manager to terminate...");
						Thread.sleep(1000);
						counter++;
						if (counter >= 10) {
							basiclog.info("Forcing Qualification Manager to terminate...");
							qmthread.interrupt();
							QualificationManager.setTerminated(true);
							break;
						}
					}
					perflog.info("Qualification Manager terminated.");
					// stop TimeConstraintsManager
					if (!isCorrSim && !isCorrSimInput && !isSim) {
						basiclog.info("Stopping Performance Plugin.");
						TimeConstraintsManager.setRunning(false);
						perflog.info("Waiting for Performance Plugin to terminate...");
						counter = 0;
						while (!TimeConstraintsManager.isTerminated()) {
							perflog.info("Still waiting for Performance Plugin to terminate...");
							Thread.sleep(1000);
							counter++;
							if (counter >= 10) {
								basiclog.info("Forcing Performance Plugin to terminate...");
								timeconstraintsthread.interrupt();
								TimeConstraintsManager.setTerminated(true);
								break;
							}
						}
						perflog.info("Performance Plugin terminated.");
					}
					// stop CSP1 etc.
					basiclog.info("Stopping Correctness Plugin.");
					AbstractQualityControlImplementation.setRunning(false);
					perflog.info("Waiting for Correctness Plugin to terminate...");
					counter = 0;
					while (!AbstractQualityControlImplementation.isTerminated()) {
						perflog.info("Still waiting for Correctness Plugin to terminate...");
						Thread.sleep(1000);
						counter++;
						if (counter >= 10) {
							basiclog.info("Forcing Correctness Plugin to terminate...");
							correctnessthread.interrupt();
							AbstractQualityControlImplementation
									.setTerminated(true);
							break;
						}
					}
					perflog.info("Correctness Plugin terminated.");

					// finalize and persist state
					Finalizer.finalizeAndPersistState(hits);
					System.exit(0);
				}
				Thread.sleep(5000);
				
				double progress = 100 * r.getCompletedAssignments()
						/ r.getRequestedAssignments();
				if (waitLongerCounter == 0)
					perflog.info("Progress: "
							+ percentageFormat.format(progress) + "%");
				else
					perflog.info("Progress: Has been a constant 100% for "
							+ (waitLongerCounter * 5 + 5)
							+ " seconds. Waiting a little longer.");
				if (waitLongerCounter == 2) {
					// it's time to stop the QualificationManager
					QualificationManager.setRunning(false);
				}
				
				boolean prepareExit = false;
				
				basiclog.debug("HITs: " + hits.length + ", Finished HITs: " + ResultOutputCollector.getCompletedHits().length + ", Escalated: " + ResultOutputCollector.getEscalatedHits().length);
				if(StaticInformation.scenario.getConnectionMode() == ConnectionMode.SIM && StaticInformation.scenario.getPrefix() == "AddressResearch")
					prepareExit = (hits.length == ResultOutputCollector.getCompletedHits().length + ResultOutputCollector.getEscalatedHits().length);
				else
					prepareExit = (r.getCompletedAssignments() >= r.getRequestedAssignments());
					
				if(prepareExit)
					waitLongerCounter++;
				else if (waitLongerCounter > 0)
					waitLongerCounter = 0;
				if (waitLongerCounter >= 4) {
					// we have waited an additional 20 seconds
				
					basiclog.info("All tasks completed.");
					basiclog.info(steps[stepCounter++]);
					// finalize and persist state
					Finalizer.finalizeAndPersistState(hits);
					HITResultFetcher.getInstance().quitHitResultFetcher();
					basiclog.info("Done!");
					System.exit(0);
				}
			}
		} catch (Exception e) {
			basiclog.error("Caught exception " + e + " during "
					+ steps[(stepCounter - 1)]);
			System.exit(-1);
		}

	}

	/**
	 * initializes the worker pools
	 * 
	 * @throws Exception
	 */
	private static void initWorkerPools() throws Exception {
		WorkerPool.createNewWorkerPool("ACTIVE", 0);
		WorkerPool.createNewWorkerPool("PENDING", 0);
		WorkerPool.createNewWorkerPool("UNKNOWN", 0);
		WorkerPool.createNewWorkerPool("BLOCKED", 0);
	}

	/**
	 * @return the quit
	 */
	public static boolean isQuit() {
		return quit;
	}

	/**
	 * @param quit
	 *            the quit to set
	 */
	public static void setQuit(boolean quit) {
		StandaloneInitializer.quit = quit;
	}

	/**
	 * @return the starting
	 */
	public static boolean isStarting() {
		return starting;
	}

}
